package cn.xiongxf.delay;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws Exception {
        // 1.创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delay_sync_consumer_group");
        // 2.设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
        // 3.订阅主题
        consumer.subscribe("Topic_delay","Part_1");
        // 4.监听消息（注册回调实现类来处理从broker拉取回来的消息）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 在处理业务的时候，报错了就返回失败，没报错就返回签收成功
                try {
                    System.out.println("签收消息成功！");
                    for (MessageExt msg : msgs) {
                        byte[] msgBody = msg.getBody();
                        System.out.println(new String(msgBody, RemotingHelper.DEFAULT_CHARSET));
                    }
                    // 标记该消息已经被成功消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    // 标记该消息未成功消费，稍后重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
        // 5.启动消费者对象（先监听，再启动）
        consumer.start();
        System.out.println("消费者启动。。。");
    }
}
